package com.bst.datax.executor.service.jobhandler;

import com.bst.common.utils.StringUtils;
import com.bst.common.utils.sql.SqlUtil;
import com.bst.datatx.core.biz.model.JobParam;
import com.bst.datatx.core.log.JobLogger;
import com.bst.datax.executor.service.logparse.LogStatistics;

import java.sql.*;
import java.time.LocalDate;
import java.util.concurrent.TimeUnit;

//@Component
//@StepScope
public class DataBatchInsertHandler {

    private String insertSql;
    private String querySql;
    private final int commitInterval=20000; // 提交间隔

    private JobParam jobParam;
    private LogStatistics logStatistics;
    public DataBatchInsertHandler(){

    }
    public DataBatchInsertHandler(JobParam jobParam,LogStatistics statistics) {
        this.jobParam = jobParam;
        this.logStatistics = statistics;
        String dayBegin = jobParam.getDayBegin();
        String dayEnd = jobParam.getDayEnd();
        insertSql = jobParam.getInsertSql();
        querySql = SqlUtil.buildSql(jobParam.getQuerySql(),dayBegin,dayEnd,"sqlserver");
        JobLogger.log("执行sql：" + insertSql+' ' + querySql);
    }

    public void executeAll(Connection conn)  throws Exception{
        java.util.Date begin = new java.util.Date();
        int i = insertSql.indexOf("values");
        insertSql = insertSql.substring(0,i);
        String insertSelectSql = insertSql+" "+querySql;
        JobLogger.log("<br>执行全量插入：" + insertSelectSql);
        int total = 0;
        try(Statement stmt = conn.createStatement();) {
            total = stmt.executeUpdate(insertSelectSql);
        } finally {
            logStatistics.setTaskRecordReaderNum(total);
            logStatistics.setTaskRecordWriteFailNum(0);
        }
        JobLogger.log("<br>全部插入，耗时" +(new java.util.Date().getTime()-begin.getTime())/1000+"s");
    }


    public void execute(Connection conn) throws Exception {
        if(StringUtils.isEmpty(insertSql) || StringUtils.isEmpty(querySql)) {
            return;
        }
        int batchSize = 0,writeSize=0;
        try (Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(querySql)) {
            // 1. 配置批量插入参数
            conn.setAutoCommit(false);
            DatabaseMetaData dbMeta = conn.getMetaData();
            JobLogger.log("<br>Batch支持情况：" + dbMeta.supportsBatchUpdates());
            try (PreparedStatement pstmt = conn.prepareStatement(insertSql)) {
                stmt.setFetchSize(50000);  // 增大游标获取量
                stmt.setFetchDirection(ResultSet.FETCH_FORWARD);
                ResultSetMetaData meta = rs.getMetaData();
                int columnCount = meta.getColumnCount();

                // 2. 使用TRY-WITH-RESOURCES自动关闭资源
                while (rs.next()) {
                    for (int i = 1; i <= columnCount; i++) {
                        pstmt.setObject(i, rs.getObject(i));
                    }
                    pstmt.addBatch();
                    // 3. 动态调整批处理提交频率
                    if (++batchSize % 5000 == 0) {  // 缩小批处理量
                        pstmt.executeBatch();
                        if (batchSize % 20000 == 0) {
                            JobLogger.log("已提交：" + batchSize + "条");
                        }
                    }
                }
                // 4. 最终提交
                pstmt.executeBatch();
                conn.commit();
                JobLogger.log("总计提交：" + batchSize + "条");
            }
        } finally {
            conn.setAutoCommit(true);  // 恢复自动提交
            logStatistics.setTaskRecordReaderNum(batchSize);
            logStatistics.setTaskRecordWriteFailNum(0);
        }
    }

    public static void disableIndexes(Connection conn, String table,String dbType) throws SQLException {
        JobLogger.log("<br>临时清除索引");
        try (Statement stmt = conn.createStatement()) {
            if(dbType.contains("SQLSERVER")) {
                stmt.execute("ALTER INDEX  ALL ON " + table + " REBUILD");
            } else {
                stmt.execute("ALTER TABLE " + table + " DISABLE KEYS");
            }
            JobLogger.log("<br>索引已清楚");
        }
    }

    public static void enableIndexes(Connection conn, String table,String dbType) throws SQLException {
        JobLogger.log("<br>恢复索引");
        try (Statement stmt = conn.createStatement()) {
            if(dbType.contains("SQLSERVER")) {
                stmt.execute("ALTER INDEX  ALL ON " + table + " ENABLE");
            } else {
                stmt.execute("ALTER TABLE " + table + " ENABLE KEYS");
            }
            JobLogger.log("<br>索引已恢复");
        }
    }

    //private static final  int batchSize =  1000;
    public  void batchDelete(Connection conn,String tableName,String dbType,String dayBegin,String dayEnd) {
        int batchSize =  1000;
        String sqlTemplate = dbType.contains("SQLSERVER")
                ? "DELETE TOP "+batchSize+" FROM "+tableName+" WITH (ROWLOCK, READPAST) WHERE ID_DIM_DAY_OCCUR BETWEEN ? AND ?"
                : "DELETE FROM "+tableName+" WHERE ID_DIM_DAY_OCCUR BETWEEN ? AND ? ";
        JobLogger.log(dbType+"删除模板为："+ sqlTemplate);
        try(PreparedStatement pstmt = conn.prepareStatement(
                sqlTemplate);) {
            pstmt.setString(1, dayBegin);
            pstmt.setString(2, dayEnd);
            int num = pstmt.executeUpdate(),total=0,batchCount=0;
            JobLogger.log(String.format("已删除 %d 条记录。", num));
            while (num > 0) {
                total += num;
                if (++batchCount % 20 == 0) {
                    JobLogger.log(String.format("已删除 %d 条记录。", total));
                }
                num = pstmt.executeUpdate();
                TimeUnit.MILLISECONDS.sleep(50);
            }
            JobLogger.log(String.format("删除完成，总计 %d 条记录", total));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
